package com.dshore.messagecenter.quartz.dataStat;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.dshore.messagecenter.model.msgrecord.PerformanceCount;
import com.dshore.messagecenter.util.UUIDUtil;

@Component
public class MonitorDateStat  implements ApplicationRunner{

	@Autowired
	private AmqpTemplate amqpTemplate;
	
	@Autowired
	private ConnectionFactory connectionFactory;
	
	// 监控数据消息队列
	private String monitorDataQueue = "monitordata.queue";
	
	@Override
	public void run(ApplicationArguments args) {
		RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
		// 启动后判断异常消息队列是否存在，不存在则新建
		if (rabbitAdmin.getQueueProperties(monitorDataQueue) == null) {
			Map<String, Object> arguments = new HashMap<String, Object>();
			arguments.put("x-max-priority", 1);
			rabbitAdmin.declareQueue(new Queue(monitorDataQueue,true, false, false, arguments));
		}
	}
	
	/**
	 * 接收异常消息
	 */
	public List<Object[]> receiveMonitorData() {
		// 获取当前时间往前一个统计节点的数据，防止与正在插入的数据产生冲突
		Calendar now = Calendar.getInstance();
		now.add(Calendar.MINUTE, -(now.get(Calendar.MINUTE) % 5));
		now.set(Calendar.SECOND, 0);
		
		// 异常消息集合
		Map<String,PerformanceCount> performanceCountMap = new HashMap<String,PerformanceCount>();
		
		// 每次调用取500条消息
		while (true) {
			// 循环读取
			Message message = amqpTemplate.receive(monitorDataQueue);
			if (message != null) {
				System.out.println(new String(message.getBody()));
				List<PerformanceCount> performanceCounts = JSON.parseArray(new String(message.getBody()), PerformanceCount.class);
				for (PerformanceCount performanceCount : performanceCounts) {
					if (!performanceCountMap.containsKey(performanceCount.getTopicQueueNo())) {
						performanceCountMap.put(performanceCount.getTopicQueueNo(), new PerformanceCount());
					}
					
					PerformanceCount record = performanceCountMap.get(performanceCount.getTopicQueueNo());
					record.setDelayTimeCollection(record.getDelayTimeCollection() + performanceCount.getDelayTimeCollection());
					record.setMsgNum(record.getMsgNum() + performanceCount.getMsgNum());
				}
			} else {
				break;
			}
		}
		
		// 当集合中有数据时，进行保存
		if (performanceCountMap.size() > 0) {
			
			List<Object[]> records = new ArrayList<Object[]>();
			
			Iterator<Entry<String, PerformanceCount>> it = performanceCountMap.entrySet().iterator();
			while (it.hasNext()) {
				Entry<String, PerformanceCount> entity = it.next();
				if (entity.getValue().getMsgNum() > 0) {
					records.add(new Object[]{UUIDUtil.getUUId(), "消息流量", entity.getValue().getMsgNum(), "主题队列", entity.getKey(), now.getTime(), now.getTime()});
					records.add(new Object[]{UUIDUtil.getUUId(), "消息发送延迟", entity.getValue().getDelayTimeCollection() / entity.getValue().getMsgNum(), "主题队列", entity.getKey(), now.getTime(), now.getTime()});
				}
			}
			return records;
		} else {
			return null;
		}
	}
}
